package com.gejia.qixishare.rocketmq.producer;

import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.UnsupportedEncodingException;
import java.util.List;

@Service
public class QixiShareMessageProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Value("${rocketmq.topic}")
    private String topic;

    @Value("${rocketmq.statisticTag}")
    private String tag;

    /**
     * 按顺序发送七夕统计数据，防止乱序
     */
    public void sendQixiShareStatisticDataByOrder(String msgContent) throws UnsupportedEncodingException {
        rocketMQTemplate.setMessageQueueSelector((List<MessageQueue> mqs, Message msg, Object arg)->{
            /**
             * mqs：要发送消息的topic下的所有消息队列集合
             * msg：发送的消息
             * arg：发送消息时传递的参数 通过该参数指定发送到哪个队列
             */
            int queueNum = Integer.valueOf(String.valueOf(arg)) % mqs.size();
            System.out.println("队列id："+queueNum+" 消息:"+new String(msg.getBody()));
            return mqs.get(queueNum);
        });
        rocketMQTemplate.syncSendOrderly(this.topic + this.tag, msgContent,"1");
    }
}
